tmsoft_dbase 0.0.7

use std::collections::HashMap;
use std::fs::{File,OpenOptions};
use std::io::{BufReader, BufWriter, Read, Write};
use std::sync::mpsc;
use std::thread;
use encoding_rs::GBK;

use memmap2::{Mmap, MmapMut};
use std::sync::Arc;
use rayon::prelude::*;
use rayon::slice::ParallelSlice;
use std::collections::HashSet;
use chrono::{Local, Datelike}; 
use anyhow::{Result, Context};
use anyhow::anyhow;
use std::fs;




#[derive(Clone)]
pub struct DbfHeader {
    pub record_count: u32,
    pub header_len: u16,
    pub record_len: u16,
    pub fields: Vec<FieldDescriptor>,
    pub file_type: u8,
    pub year: u8,
    pub month: u8,
    pub day: u8,
    pub reserved: [u8; 20],  // 20 字节的保留数据
}

#[derive(Clone)]
#[derive(Debug)]
pub struct FieldDescriptor {
    pub name: String,
    pub field_type: char,
    pub length: u8,
    pub decimal: u8,
}

#[derive(Debug, Clone)]
struct FieldDescriptor1 {
    name: String,
    field_type: u8, // 修改为 u8 类型
    length: u8,
    decimal: u8,
}
/// 解析单个 DBF 记录。
///
/// # 参数
///
/// * `data` - 表示记录数据的字节切片。
/// * `fields` - 表示字段描述符的 `FieldDescriptor` 切片。
///
/// # 返回值
///
/// 一个 `HashMap`，其中键是字段名，值是字段值。
/// 带缓存的GBK解码器
/// 缓存 GBK 解码器
// 定义一个全局的解码器实例用于缓存
/// 解析记录函数
// 终极优化版解析
fn parse_record(data: &[u8], fields: &[FieldDescriptor]) -> Result<HashMap<String, String>> {
    let mut record = HashMap::new();
    let mut offset = 1; // Skip the deletion flag (assuming first byte is the deletion flag)

    for field in fields {
        let end = offset + field.length as usize;

        // Prevent out-of-bounds access
        if end > data.len() {
            break;
        }

        // Get the field data slice
        let field_data = &data[offset..end];

        // Decode the field using GBK and handle the Result
        let (value, _, _) = GBK.decode(field_data);
        let value = value.trim().to_string();

        // Insert the value into the record
        record.insert(field.name.clone(), value);

        // Update the offset for the next field
        offset = end;
    }

    Ok(record)
}



/// 优化版记录解析（复用字段位置信息）
// fn parse_record(data: &[u8], fields: &[FieldDescriptor]) -> HashMap<String, String> {
//     let mut pos = 1;
//     fields.iter()
//         .map(|field| {
//             let end = pos + field.length as usize;
//             let value = String::from_utf8_lossy(&data[pos..end])
//                 .trim_end().to_string();
//             pos = end;
//             (field.name.clone(), value)
//         })
//         .collect()
// }



pub fn read_dbf_concurrent(path: &str) -> Result<Vec<HashMap<String, String>>> {
    let file = File::open(path).context("Failed to open DBF file")?;
    let mmap = unsafe { Mmap::map(&file).context("Failed to map DBF file into memory")? };
    
    // 获取完整头部数据（假设header_len不超过映射长度）
    let header = parse_header(&mmap)?;

    let num_threads = num_cpus::get().min(header.record_count as usize / 1000 + 1);
    let records_per_thread = (header.record_count as usize + num_threads - 1) / num_threads;
    
    let mmap = Arc::new(mmap); // 共享内存映射
    let (tx, rx) = mpsc::channel();
    
    let record_size = header.record_len as usize;
    let base_offset = header.header_len as usize;

    for i in 0..num_threads {
        let tx = tx.clone();
        let mmap = Arc::clone(&mmap);
        let fields = header.fields.clone();  // 确保传递的字段描述符是最新的
        
        thread::spawn(move || {
            let start_idx = i * records_per_thread;
            let end_idx = (start_idx + records_per_thread).min(header.record_count as usize);
            
            let start_offset = base_offset + start_idx * record_size;
            let end_offset = base_offset + end_idx * record_size;
            
            let chunk = &mmap[start_offset..end_offset];
            let mut records = Vec::with_capacity(end_idx - start_idx);
            
            for (idx, record_data) in chunk.chunks(record_size).enumerate() {
                if record_data[0] == 0x2A { continue; } // 跳过删除标记
                let _global_idx = start_idx + idx;

                // Here we propagate any error from `parse_record` using the `?` operator
                match parse_record(record_data, &fields) {
                    Ok(record) => records.push(record),
                    Err(e) => {
                        eprintln!("Error parsing record: {:?}", e);
                    }
                }
            }
            
            tx.send(records).unwrap();
        });
    }

    drop(tx); // Close the sender
    let mut all_records = Vec::with_capacity(header.record_count as usize);
    for received in rx {
        all_records.extend(received);
    }

    Ok(all_records)
}


/// 从给定的文件解析 DBF 头部。
///
/// # 参数
///
/// * `file` - 要读取头部的文件的可变引用。
///
/// # 返回值
///
/// 一个包含解析后的 `DbfHeader` 或错误的 `Result`。
/// 


// 原始版本（基于File）
// fn parse_header(file: &mut File) -> Result<DbfHeader, Box<dyn std::error::Error>> {
//     let mut header_data = [0u8; 32];
//     file.read_exact(&mut header_data)?;

//     let record_count = u32::from_le_bytes(header_data[4..8].try_into()?);
//     let header_len = u16::from_le_bytes(header_data[8..10].try_into()?);
//     let record_len = u16::from_le_bytes(header_data[10..12].try_into()?);

//     let fields = parse_field_descriptors(file, header_len)?;

//     Ok(DbfHeader {
//         record_count,
//         header_len,
//         record_len,
//         fields,
//     })
// }

// 改造后版本（基于字节切片）
// 仍然使用 &[u8] 作为输入，保留原始的解析逻辑
fn parse_header(data: &[u8]) -> Result<DbfHeader> {
    // 确保有足够的数据读取基础头
    if data.len() < 32 {
        return Err(anyhow::anyhow!("Incomplete header data").into());
    }

    // 直接读取前32字节
    let header_data = &data[0..32];

    // 保持原有解析逻辑
    let record_count = u32::from_le_bytes(header_data[4..8].try_into()?);
    let header_len = u16::from_le_bytes(header_data[8..10].try_into()?);
    let record_len = u16::from_le_bytes(header_data[10..12].try_into()?);

    // 读取 reserved 字段（20 字节）
    let reserved = header_data[12..32].try_into()?;

    // 计算字段描述符范围
    let fields_start = 32;  // 基础头之后
    let fields_end = header_len as usize;  // 总头长度
    
    // 验证数据范围
    if data.len() < fields_end {
        return Err(anyhow::anyhow!("Field descriptors exceed data bounds").into());
    }

    // 改造字段解析函数
    let fields = parse_field_descriptors(&data[fields_start..fields_end])?;

    // 返回 DbfHeader
    Ok(DbfHeader { 
        record_count, 
        header_len, 
        record_len, 
        fields, 
        reserved, // 新增的 reserved 字段
        file_type: header_data[0], // 文件类型在 header_data 的第一个字节
        year: header_data[1], // 最后更新年
        month: header_data[2], // 最后更新月
        day: header_data[3], // 最后更新日
    })
}






/// 从给定的文件解析字段描述符。
///
/// # 参数
///
/// * `file` - 要读取字段描述符的文件的可变引用。
/// * `header_len` - 头部的长度。
///
/// # 返回值
///
/// 一个包含 `FieldDescriptor` 向量或错误的 `Result`。

fn parse_field_descriptors(data: &[u8]) -> Result<Vec<FieldDescriptor>> {
    let mut fields = Vec::new();
    let mut cursor = 0;

    // 每32字节为一个字段描述符
    while cursor + 32 <= data.len() {
        let chunk = &data[cursor..cursor+32];
        
        // 字段名（0-10字节）
        let name = String::from_utf8_lossy(&chunk[0..11])
            .trim_end_matches('\0')
            .to_string();

        // 字段类型（11字节）
        let field_type = char::from(chunk[11]);

        // 字段长度（16字节）和小数位数（17字节）
        let length = chunk[16];  // 直接读取u8
        let decimal = chunk[17]; // 新增小数位数解析

        fields.push(FieldDescriptor {
            name,
            field_type,
            length,
            decimal, // 添加缺失的字段
        });

        cursor += 32;

        // 检查终止符（0x0D）
        if cursor < data.len() && data[cursor] == 0x0D {
            break;
        }
    }

    Ok(fields)
}

/// 从给定的文件路径获取 DBF 文件的字段信息。
///
/// # 参数
///
/// * `path` - DBF 文件的路径。
///
/// # 返回值
///
/// 一个包含字段描述符向量或错误的 `Result`。
/// 从给定的文件路径获取 DBF 文件的字段信息
pub fn get_dbf_fields(path: &str) -> Result<Vec<FieldDescriptor>> {
    // 打开文件并创建内存映射
    let file = File::open(path).context("Failed to open DBF file")?;
    let mmap = unsafe { Mmap::map(&file).context("Failed to map DBF file into memory")? };
    
    // 直接使用完整内存映射数据解析头部
    let header = parse_header(&mmap)?;

    Ok(header.fields)
}

/// 查找 DBF 文件中的记录。
///
/// # 参数
///
/// * `path` - DBF 文件的路径。
/// * `search_fn` - 一个闭包，用于查找记录。
///
/// # 返回值
///
/// 一个包含找到的记录或错误的 `Result`。
pub fn find_records(
    path: &str,
    query: &HashMap<String, String>,
) -> Result<Vec<HashMap<String, String>>> {
    let file = File::open(path).context("Failed to open DBF file")?;
    let mmap = unsafe { Mmap::map(&file).context("Failed to map DBF file into memory")? };

    let header = parse_header(&mmap)?;
    let num_threads = num_cpus::get().min(header.record_count as usize / 1000 + 1);
    let records_per_thread = (header.record_count as usize + num_threads - 1) / num_threads;

    let mmap = Arc::new(mmap); // 共享内存映射
    let (tx, rx) = mpsc::channel();

    let record_size = header.record_len as usize;
    let base_offset = header.header_len as usize;
    let fields = Arc::new(header.fields);

    for i in 0..num_threads {
        let tx = tx.clone();
        let mmap = Arc::clone(&mmap);
        let fields = Arc::clone(&fields);
        let query = Arc::new(query.clone());

        thread::spawn(move || {
            let start_idx = i * records_per_thread;
            let end_idx = (start_idx + records_per_thread).min(header.record_count as usize);

            let start_offset = base_offset + start_idx * record_size;
            let end_offset = base_offset + end_idx * record_size;

            let chunk = &mmap[start_offset..end_offset];
            let mut records = Vec::with_capacity(end_idx - start_idx);

            for (idx, record_data) in chunk.chunks(record_size).enumerate() {
                if record_data[0] == 0x2A { continue; } // 跳过删除标记

                match parse_record(record_data, &fields) {
                    Ok(record) => {
                        // 检查是否满足查询条件
                        if record_matches_query(&record, &query) {
                            records.push(record);
                        }
                    }
                    Err(e) => {
                        eprintln!("Error parsing record: {:?}", e);
                    }
                }
            }

            tx.send(records).unwrap();
        });
    }

    drop(tx); // 关闭发送端
    let mut all_records = Vec::new();

    for received in rx {
        all_records.extend(received);
    }

    Ok(all_records)
}

fn record_matches_query(
    record: &HashMap<String, String>,
    query: &HashMap<String, String>,
) -> bool {
    query.iter().all(|(key, value)| {
        record.get(key).map_or(false, |record_value| record_value == value)
    })
}



/// 高效去除尾部空格（无需字符串转换）
// fn trim_trailing_spaces(bytes: &[u8]) -> &[u8] {
//     let mut end = bytes.len();
//     while end > 0 && (bytes[end-1] == b' ' || bytes[end-1] == 0) {
//         end -= 1;
//     }
//     &bytes[..end]
// }

/// 将 DBF 记录写入给定的文件路径。
///
/// # 参数
///
/// * `path` - DBF 文件的路径。
/// * `records` - 记录的切片（每个记录是字段名到值的 `HashMap`）。
/// * `fields` - 表示字段描述符的 `FieldDescriptor` 切片。
///
/// # 返回值
///
/// 一个表示成功或失败的 `Result`。
pub fn write_dbf(path: &str, records: &[HashMap<String, String>], fields: &[FieldDescriptor]) -> Result<(), std::io::Error> {
    // 创建文件并包装为 BufWriter
    let file = File::create(path)?;
    let mut writer = BufWriter::new(file);

    // 文件头日期处理（使用 chrono）
    let today = Local::now();

    let header_len = 32 + fields.len() * 32 + 1;
    let record_len = fields.iter().map(|f| f.length as usize).sum::<usize>() + 1;

    let mut header = vec![0u8; 32];
    header[0] = 0x03;  // DBF 文件版本号
    header[1] = (today.year() - 1900) as u8;  // 年份，减去1900
    header[2] = today.month() as u8;  // 月份
    header[3] = today.day() as u8;  // 日期
    header[4..8].copy_from_slice(&(records.len() as u32).to_le_bytes());
    header[8..10].copy_from_slice(&(header_len as u16).to_le_bytes());
    header[10..12].copy_from_slice(&(record_len as u16).to_le_bytes());

    writer.write_all(&header)?;

    // 写入字段描述符
    for field in fields {
        let mut field_data = [0u8; 32];
        let name_bytes = field.name.as_bytes();
        let name_len = name_bytes.len().min(11); // 字段名称最大长度为 11
        field_data[0..name_len].copy_from_slice(&name_bytes[0..name_len]);
        field_data[11] = field.field_type as u8;  // 字段类型
        field_data[16] = field.length;  // 字段长度
        field_data[17] = field.decimal;  // 小数位数
        writer.write_all(&field_data)?;
    }

    writer.write_all(&[0x0D])?; // 结束符，表示字段描述结束

    // 处理每条记录
    let mut buffer = Vec::with_capacity(record_len);
    for record in records {
        buffer.clear();
        buffer.push(0x20);  // 默认标记为未删除（0x20 是空格字符）

        for field in fields {
            let value = record.get(&field.name).map(|s| s.as_str()).unwrap_or("");  // 获取字段值，默认值为空字符串
            let encoded = if field.field_type == 'C' {
                // 对字符字段进行 GBK 编码
                let (encoded, _, _) = GBK.encode(value);
                encoded.to_vec()
            } else {
                // 对数值字段直接转化为字节数组
                value.as_bytes().to_vec()
            };

            // 填充字段数据
            let mut padded = vec![0x20; field.length as usize];  // 默认用空格填充
            let copy_len = encoded.len().min(field.length as usize);  // 字段数据不能超过字段定义的长度

            match field.field_type {
                'C' => {
                    padded[..copy_len].copy_from_slice(&encoded[..copy_len]);  // 对字符字段进行拷贝
                },
                _ => {
                    let start = field.length as usize - copy_len;  // 数值字段填充对齐
                    padded[start..].copy_from_slice(&encoded[..copy_len]);
                }
            }
            buffer.extend(&padded);  // 将填充后的数据写入 buffer
        }
        writer.write_all(&buffer)?;
    }

    writer.flush()?;
    Ok(())
}


/// 更新符合条件的 DBF 记录。
///
/// # 参数
///
/// * `path` - DBF 文件的路径。
/// * `query` - 查询条件。
/// * `updates` - 更新内容。
///
/// # 返回值
///
/// 一个包含更新记录数或错误的 `Result`。
pub fn update_records(path: &str, query: HashMap<String, String>, updates: HashMap<String, String>) -> Result<usize, std::io::Error> {
    // 读取所有记录
    let mut all_records = read_dbf_concurrent(path)
        .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))?;  // 将 anyhow::Error 转换为 std::io::Error
    let mut updated_count = 0;

    // 更新符合条件的记录
    for record in all_records.iter_mut() {
        let mut match_query = true;
        for (key, value) in &query {
            if let Some(record_value) = record.get(key) {
                if record_value != value {
                    match_query = false;
                    break;
                }
            } else {
                match_query = false;
                break;
            }
        }
        if match_query {
            for (key, value) in &updates {
                if let Some(record_value) = record.get_mut(key) {
                    *record_value = value.clone();
                }
            }
            updated_count += 1;
        }
    }

    // 打印更新后的所有记录
    println!("更新后的所有记录:");
    for record in &all_records {
        for (key, value) in record {
            println!("{} -> {}", key, value);
        }
        println!("------");
    }

    // 获取字段信息
    let fields = get_dbf_fields(path)
        .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))?;  // 同样转换

    // 写入更新后的记录到原始 DBF 文件
    write_dbf(path, &all_records, &fields)?;

    Ok(updated_count)
}



/// 增加新的 DBF 记录。
///
/// # 参数
///
/// * `path` - DBF 文件的路径。
/// * `new_record` - 新记录（字段名到值的 `HashMap`）。
///
/// # 返回值
///
/// 一个表示成功或失败的 `Result`。
pub fn add_record(path: &str, mut new_record: HashMap<String, String>) -> Result<(), std::io::Error> {
    // 读取所有记录
    let mut all_records = read_dbf_concurrent(path)
        .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))?;  // 将 anyhow::Error 转换为 std::io::Error

    // 获取字段信息
    let fields = get_dbf_fields(path)
        .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))?;  // 同样转换

    // 检查新记录是否包含所有字段，如果没有则补上默认值
    for field in &fields {
        if !new_record.contains_key(&field.name) {
            new_record.insert(field.name.clone(), String::new());
        }
    }

    // 检查新记录是否符合字段的约束条件
    for field in &fields {
        if let Some(value) = new_record.get(&field.name) {
            match field.field_type {
                'C' => {
                    // 字符型字段长度检查
                    if value.len() > field.length as usize {
                        return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, format!("Field {} exceeds maximum length of {}", field.name, field.length)));
                    }
                }
                'N' => {
                    // 数值型字段格式检查
                    if value.parse::<f64>().is_err() {
                        return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, format!("Field {} must be a valid number", field.name)));
                    }
                }
                _ => {}
            }
        }
    }

    // 增加新记录
    all_records.push(new_record.clone());

    // 打印新增记录
    println!("新增记录:");
    for (key, value) in &new_record {
        println!("{} -> {}", key, value);
    }

    // 写入更新后的记录到原始 DBF 文件
    write_dbf(path, &all_records, &fields)
        .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))?;  // 转换写入错误

    Ok(())
}



/// 删除符合条件的 DBF 记录。
///
/// # 参数
///
/// * `path` - DBF 文件的路径。
/// * `query` - 查询条件。
///
/// # 返回值
///
/// 一个包含删除记录数或错误的 `Result`。
pub fn delete_records(path: &str, query: &HashMap<String, String>) -> Result<usize> {
    // 读取所有记录
    let mut all_records = read_dbf_concurrent(path)
        .context("Failed to read DBF records")?;
    let original_count = all_records.len();

    // 过滤掉符合条件的记录
    all_records.retain(|record| {
        for (key, value) in query {
            if let Some(record_value) = record.get(key) {
                if record_value != value {
                    return true;
                }
            } else {
                return true;
            }
        }
        false
    });

    let deleted_count = original_count - all_records.len();

    // 获取字段信息
    let fields = get_dbf_fields(path)
        .context("Failed to get DBF fields")?;

    // 写入更新后的记录到原始 DBF 文件
    write_dbf(path, &all_records, &fields)
        .context("Failed to write updated DBF file")?;

    Ok(deleted_count)
}

/// 获取 DBF 文件的总记录数。
///
/// # 参数
///
/// * `path` - DBF 文件的路径。
///
/// # 返回值
///
/// 一个包含记录数或错误的 `Result`。
/// 获取 DBF 文件的总记录数
pub fn get_record_count(path: &str) -> Result<u32> {
    // 使用内存映射优化读取
    let file = File::open(path)
        .context(format!("Failed to open file: {}", path))?;
    let mmap = unsafe { Mmap::map(&file) }
        .context("Failed to map file to memory")?;
    
    // 添加基础校验（确保有足够的数据读取记录数）
    if mmap.len() < 8 { // 记录数位于偏移量4-8字节
        return Err(anyhow::anyhow!("文件太小无法读取有效头部").into());
    }

    // 直接解析记录数（可选优化方案）
    let record_count = u32::from_le_bytes(mmap[4..8].try_into()?);
    
    Ok(record_count)
}

/// 从给定的文件路径分页读取 DBF 记录。
///
/// # 参数
///
/// * `path` - DBF 文件的路径。
/// * `page` - 页码，从 1 开始。
/// * `page_size` - 每页的记录数。
///
/// # 返回值
///
/// 一个包含记录向量（每个记录是字段名到值的 `HashMap`）或错误的 `Result`。
/// 分页读取DBF记录（内存映射+多线程解析）
pub fn read_dbf_paged(
    path: &str,
    page: usize,
    page_size: usize,
) -> Result<Vec<HashMap<String, String>>> {
    // 打开文件并内存映射
    let file = File::open(path)
        .context(format!("Failed to open file: {}", path))?;
    let mmap = Arc::new(unsafe { Mmap::map(&file) }
        .context("Failed to map file to memory")?);
    let header = parse_header(&mmap)
        .context("Failed to parse DBF header")?;

    // 计算记录范围
    let total_records = header.record_count as usize;
    let start_idx = (page - 1).saturating_mul(page_size);
    let end_idx = (start_idx + page_size).min(total_records);

    // 处理超出范围的情况
    if start_idx >= total_records {
        return Ok(Vec::new());
    }

    // 计算字节范围
    let record_size = header.record_len as usize;
    let data_start = header.header_len as usize;
    let start_offset = data_start + start_idx * record_size;
    let end_offset = data_start + end_idx * record_size;

    // 获取记录数据切片
    let records_data = &mmap[start_offset..end_offset.min(mmap.len())];

    // 并行解析记录
    let records: Result<Vec<_>, _> = records_data
        .par_chunks(record_size)
        .filter(|chunk| chunk[0] != 0x2A)
        .map(|chunk| parse_record(chunk, &header.fields))
        .collect();

    records.map_err(|e| anyhow::anyhow!("Error parsing records: {}", e).into())
}


/// 从给定的文件路径并发读取 DBF 记录（带查询条件和分页）
///
/// # 参数
///
/// * `path` - DBF 文件的路径
/// * `query` - 查询条件（字段名到值的映射）
/// * `page` - 页码（从 1 开始）
/// * `page_size` - 每页记录数
///
/// # 返回值
///
/// 包含分页记录向量或错误的 Result
pub fn read_dbf_paged_query(
    path: &str,
    query: &HashMap<String, String>,
    page: usize,
    page_size: usize,
) -> Result<Vec<HashMap<String, String>>> {
    let file = File::open(path).context("Failed to open DBF file")?;
    let mmap = unsafe { Mmap::map(&file) }.context("Failed to memory-map DBF file")?;
    let header = parse_header(&mmap).context("Failed to parse DBF header")?;

    let num_threads = num_cpus::get().min(header.record_count as usize / 1000 + 1);
    let records_per_thread = (header.record_count as usize + num_threads - 1) / num_threads;

    let mmap = Arc::new(mmap);
    let query = Arc::new(query.clone());
    let (tx, rx) = mpsc::channel();

    let record_size = header.record_len as usize;
    let base_offset = header.header_len as usize;

    for i in 0..num_threads {
        let tx = tx.clone();
        let mmap = Arc::clone(&mmap);
        let fields = header.fields.clone();
        let query = Arc::clone(&query);

        thread::spawn(move || {
            let start_idx = i * records_per_thread;
            let end_idx = (start_idx + records_per_thread).min(header.record_count as usize);

            let start_offset = base_offset + start_idx * record_size;
            let end_offset = base_offset + end_idx * record_size;

            let chunk = &mmap[start_offset..end_offset];
            let mut records = Vec::with_capacity(end_idx - start_idx);

            for record_data in chunk.chunks(record_size) {
                if record_data[0] == 0x2A {
                    continue;
                }

                // Error handling inside thread
                match parse_record(record_data, &fields) {
                    Ok(record) => {
                        if matches_query(&record, &query) {
                            records.push(record);
                        }
                    }
                    Err(e) => {
                        eprintln!("Error parsing record: {}", e);
                    }
                }
            }

            tx.send(records).unwrap();
        });
    }

    drop(tx);
    let mut all_records = Vec::new();
    for received in rx {
        all_records.extend(received);
    }

    // Pagination handling
    let start = (page - 1) * page_size;
    let end = start + page_size;
    let total = all_records.len();

    Ok(if start >= total {
        Vec::new()
    } else {
        all_records[start..end.min(total)].to_vec()
    })
}

/// 检查记录是否匹配查询条件
fn matches_query(record: &HashMap<String, String>, query: &HashMap<String, String>) -> bool {
    query.iter()
        .all(|(k, v)| record.get(k).map_or(false, |val| val == v))
}

pub fn add_fields_to_dbf(path: &str, add_fields: Vec<FieldDescriptor>) -> Result<()> {
    // 读取原文件全部内容
    let  file_data = {
        let file = File::open(path).context("Failed to open DBF file")?;
        let mut buf = Vec::new();
        BufReader::new(file).read_to_end(&mut buf).context("Failed to read DBF file")?;
        buf
    };

    // 解析原头部
    let header = parse_header(&file_data).context("Failed to parse DBF header")?;

    // 验证原始记录长度
    if header.record_len == 0 {
        return Err(anyhow!("Original record length cannot be zero"));
    }
    let old_record_len = header.record_len as usize;

    // 检查文件数据完整性
    let data_start = header.header_len as usize;
    let expected_data_len = header.record_count as usize * old_record_len;
    if file_data.len() < data_start + expected_data_len {
        return Err(anyhow!("File data is incomplete or corrupted"));
    }

    // 检查并添加新字段
    let mut fields = header.fields.clone();
    let existing_names: HashSet<_> = fields.iter().map(|f| &f.name).collect();
    let new_fields = add_fields.into_iter()
        .filter(|f| !existing_names.contains(&f.name))
        .collect::<Vec<_>>();
    fields.extend(new_fields);

    // 计算新头部长度和记录长度
    let updated_header_len = 32 + (fields.len() * 32) as u16 + 1; // 包含终止符
    let updated_record_len = fields.iter().map(|f| f.length as usize).sum::<usize>();

    // 验证新记录长度
    if updated_record_len < old_record_len {
        return Err(anyhow!("New record length is shorter than the original record length"));
    }

    // 重建数据记录
    let mut new_data = Vec::with_capacity(header.record_count as usize * updated_record_len);

    for chunk in file_data[data_start..].chunks(old_record_len) {
        let mut record = chunk.to_vec();

        // 如果 chunk 长度不足，用空格填充到 old_record_len
        if record.len() < old_record_len {
            let padding_len = old_record_len - record.len();
            record.extend(vec![b' '; padding_len]);
        }

        // 扩展记录长度，填充新字段默认值
        let padding_len = updated_record_len - old_record_len;
        record.extend(vec![b' '; padding_len]);

        new_data.extend_from_slice(&record);
    }

    // 构建新文件内容
    let mut new_file = Vec::new();

    // 写入新头部
    // 主头部结构（共32字节）：
    // [0] 文件类型
    // [1] 最后更新年（当前年 - 1900）
    // [2] 最后更新月
    // [3] 最后更新日
    // [4-7] 记录数（小端）
    // [8-9] 头部长度（小端）
    // [10-11] 记录长度（小端）
    // [12-31] 保留字段（20字节）

    // 文件类型（1字节）
    new_file.push(header.file_type);
    // 最后更新日期（3字节）
    new_file.push(header.year);
    new_file.push(header.month);
    new_file.push(header.day);
    // 记录数（4字节小端）
    new_file.extend_from_slice(&header.record_count.to_le_bytes());
    // 头部长度（2字节小端）
    new_file.extend_from_slice(&updated_header_len.to_le_bytes());
    // 记录长度（2字节小端）
    new_file.extend_from_slice(&(updated_record_len as u16).to_le_bytes());
    // 保留字段（20字节）
    new_file.extend_from_slice(&header.reserved);

    // 写入字段描述符
    for field in &fields {
        let mut desc = [0; 32];
        let name_bytes = field.name.as_bytes();
        let len = name_bytes.len().min(11);
        desc[..len].copy_from_slice(&name_bytes[..len]);
        desc[11] = field.field_type as u8;
        desc[16] = field.length;
        desc[17] = field.decimal;
        new_file.extend_from_slice(&desc);
    }

    // 头部终止符
    new_file.push(0x0D);

    // 写入处理后的数据
    new_file.extend_from_slice(&new_data);

    // 写入文件（原子操作）
    fs::write(path, new_file).context("Failed to write updated DBF file")?;

    Ok(())
}

// 修改字段方法



// 新的 FieldDescriptor1 定义



pub fn modify_fields_in_dbf(path: &str, modify_fields: Vec<FieldDescriptor>) -> Result<()> {
    // 打开文件以进行读写操作
    let file = OpenOptions::new()
        .read(true)
        .write(true)
        .open(path)
        .context(format!("Failed to open file with write access: {}", path))?;

    // 创建可变内存映射 (使用 memmap2)
    let mut mmap_mut = unsafe { MmapMut::map_mut(&file).context("Failed to map file to memory")? };

    // 解析 DBF 头部
    let mut header = parse_header(&mmap_mut).context("Failed to parse DBF header")?;

    // 修改字段
    let mut fields = header.fields.clone();
    for modify_field in modify_fields {
        if let Some(field) = fields.iter_mut().find(|f| f.name == modify_field.name) {
            *field = modify_field;
        }
    }

    // 将原始 FieldDescriptor 转换为 FieldDescriptor1
    let fields1: Vec<FieldDescriptor1> = fields
        .iter()
        .map(|field| FieldDescriptor1 {
            name: field.name.clone(),
            field_type: field.field_type as u8, // 将 char 转换为 u8
            length: field.length,
            decimal: field.decimal,
        })
        .collect();

    // 更新头部信息
    let updated_header_len = 32 + fields1.len() as u16 * 32 + 1; // +1 for terminator
    let updated_record_len = fields1.iter().map(|f| f.length as u16).sum::<u16>();
    header.header_len = updated_header_len;
    header.record_len = updated_record_len;

    // 计算所需文件大小
    let total_records = header.record_count as usize;
    let old_record_size = header.record_len as usize;
    let new_record_size = updated_record_len as usize;
    let data_start = header.header_len as usize;
    let required_file_size = data_start + total_records * new_record_size + 1; // +1 for 0x1A

    // 如果文件大小不足，调整文件大小
    if required_file_size > mmap_mut.len() {
        println!(
            "Resizing file: old_size={}, new_size={}",
            mmap_mut.len(),
            required_file_size
        );
        file.set_len(required_file_size as u64)
            .context("Failed to resize file")?;
        // 重新创建内存映射 (使用 memmap2)
        mmap_mut = unsafe { MmapMut::map_mut(&file).context("Failed to re-map file")? };
    }

    // 更新头部信息到内存映射
    let header_data = &mut mmap_mut[0..32];
    header_data[0] = header.file_type;
    header_data[1] = header.year;
    header_data[2] = header.month;
    header_data[3] = header.day;
    header_data[4..8].copy_from_slice(&header.record_count.to_le_bytes());
    header_data[8..10].copy_from_slice(&header.header_len.to_le_bytes());
    header_data[10..12].copy_from_slice(&header.record_len.to_le_bytes());
    header_data[12..32].copy_from_slice(&header.reserved);

    // 写入字段描述符后的终止符 0x0D
    let terminator_pos = 32 + fields1.len() * 32;
    if terminator_pos >= mmap_mut.len() {
        return Err(anyhow!("Terminator position out of bounds"));
    }
    mmap_mut[terminator_pos] = 0x0D;

    // 更新字段描述符
    let fields_start = 32;
    for (i, field) in fields1.iter().enumerate() {
        let field_data_start = fields_start + i * 32;
        let field_data = &mut mmap_mut[field_data_start..field_data_start + 32];

        // 写入字段名称（GBK 编码）
        let (name_bytes, _, _) = GBK.encode(&field.name);
        let name_len = name_bytes.len().min(11); // 确保字段名称不超过 11 字节
        field_data[0..name_len].copy_from_slice(&name_bytes[..name_len]);
        field_data[name_len] = 0; // Null terminator

        // 写入字段类型、长度和小数位数
        field_data[11] = field.field_type; // 使用 u8 类型
        field_data[16] = field.length;
        field_data[17] = field.decimal;

        // 清零剩余字节
        for j in 12..32 {
            if j != 16 && j != 17 {
                field_data[j] = 0;
            }
        }
    }

    // 更新记录数据
    // 更新记录数据
let original_fields = fields1.clone(); // 使用转换后的 FieldDescriptor1
for record_idx in 0..total_records {
    let old_offset = data_start + record_idx * old_record_size;
    let new_offset = data_start + record_idx * new_record_size;

    // 预校验偏移量（安全增强）
    if old_offset >= mmap_mut.len() || new_offset >= mmap_mut.len() {
        return Err(anyhow!("Record offset out of bounds"));
    }

    // 设置删除标记（保持不变）
    mmap_mut[new_offset] = mmap_mut[old_offset];

    let mut pos_old = 1;
    let mut pos_new = 1;

    for (i, new_field) in fields1.iter().enumerate() {
        let old_field = &original_fields[i]; // 保存修改前的字段信息
        let old_data_range = pos_old..(pos_old + old_field.length as usize);
        let new_data_range = pos_new..(pos_new + new_field.length as usize);

        // 边界检查（安全增强）
        if old_offset + old_data_range.end > mmap_mut.len()
            || new_offset + new_data_range.end > mmap_mut.len()
        {
            return Err(anyhow!("Field data out of bounds"));
        }

        // 先读取旧数据到临时缓冲区
        let old_data: Vec<u8> = mmap_mut[old_offset + old_data_range.start..old_offset + old_data_range.end]
            .to_vec();

        // 创建一个临时缓冲区用于存储新数据
        let mut new_data_buffer = vec![b' '; new_field.length as usize];

        // 处理字段数据
        match new_field.field_type {
            b'C' => process_string_field(&old_data, &mut new_data_buffer, old_field, new_field),
            b'N' => process_numeric_field(&old_data, &mut new_data_buffer, old_field, new_field),
            b'D' => process_date_field(&old_data, &mut new_data_buffer),
            b'L' => process_logical_field(&old_data, &mut new_data_buffer),
            _ => new_data_buffer.copy_from_slice(&old_data),
        }

        // 将新数据写回 mmap_mut
        let new_data_start = new_offset + new_data_range.start;
        let new_data_end = new_offset + new_data_range.end;
        mmap_mut[new_data_start..new_data_end].copy_from_slice(&new_data_buffer);

        pos_old += old_field.length as usize;
        pos_new += new_field.length as usize;
    }

    // 清空多余的空间
    if new_record_size > pos_new {
        if new_offset + new_record_size > mmap_mut.len() {
            return Err(anyhow!(
                "New record size out of bounds: new_offset={}, new_record_size={}, mmap_len={}",
                new_offset, new_record_size, mmap_mut.len()
            ));
        }
        mmap_mut[new_offset + pos_new..new_offset + new_record_size].fill(b' ');
    }
}

    // 写入文件结束符 0x1A
    if required_file_size > 0 {
        mmap_mut[required_file_size - 1] = 0x1A;
    }

    // 刷新内存映射到文件
    mmap_mut.flush().context("Failed to flush memory map changes to file")?;
    file.sync_all().context("Failed to sync file changes")?;
    println!("Changes saved to file.");

    Ok(())
}

// 处理字符串字段的核心逻辑
fn process_string_field(
    old_data: &[u8],
    new_data: &mut [u8],
    old_field: &FieldDescriptor1,
    new_field: &FieldDescriptor1,
) {
    // 解码旧数据
    let (decoded, _, _) = GBK.decode(old_data);
    let decoded_str = decoded.trim_end().to_string();

    // 编码新数据
    let (encoded, _, _) = GBK.encode(&decoded_str);
    let write_len = encoded.len().min(new_data.len());

    // 写入新数据并填充
    new_data[..write_len].copy_from_slice(&encoded[..write_len]);
    new_data[write_len..].fill(b' ');
}

// 数值型字段处理
fn process_numeric_field(
    old_data: &[u8],
    new_data: &mut [u8],
    old_field: &FieldDescriptor1,
    new_field: &FieldDescriptor1,
) {
    // 直接迁移数值数据
    let copy_len = old_data.len().min(new_data.len());
    new_data[..copy_len].copy_from_slice(&old_data[..copy_len]);

    // 右对齐数值数据
    if new_data.len() > copy_len {
        let padding = new_data.len() - copy_len;
        new_data.copy_within(..copy_len, padding);
        new_data[..padding].fill(b' ');
    }
}

// 日期型字段处理 (格式: YYYYMMDD)
fn process_date_field(old_data: &[u8], new_data: &mut [u8]) {
    // 直接复制日期数据
    let copy_len = old_data.len().min(new_data.len());
    new_data[..copy_len].copy_from_slice(&old_data[..copy_len]);
}

// 逻辑型字段处理 (1字节)
fn process_logical_field(old_data: &[u8], new_data: &mut [u8]) {
    // 直接复制逻辑数据
    if !old_data.is_empty() {
        new_data[0] = old_data[0];
    }
}


// 删除字段方法


pub fn delete_fields_from_dbf(
    path: &str,
    delete_field_names: Vec<String>,
) -> Result<()> {
    // 打开文件以进行读写操作
    let file = OpenOptions::new()
        .read(true)
        .write(true)
        .open(path)
        .context("Failed to open file")?;

    // 创建可变内存映射
    let mut mmap = unsafe { MmapMut::map_mut(&file).context("Failed to map file")? };

    // 解析 DBF 头部（immutable borrow）
    let header = {
        let header = parse_header(&mmap)?;
        header // Drop the immutable borrow here
    };

    // 删除指定字段
    let mut fields = header.fields.clone();
    fields.retain(|field| !delete_field_names.contains(&field.name));

    // 更新头部信息
    let updated_header_len = 32 + fields.len() as u16 * 32;
    let updated_record_len = fields.iter().map(|f| f.length as u16).sum::<u16>();
    let updated_header = DbfHeader {
        record_count: header.record_count,
        header_len: updated_header_len,
        record_len: updated_record_len,
        fields,
        file_type: header.file_type,
        year: header.year,
        month: header.month,
        day: header.day,
        reserved: header.reserved,
    };

    // 更新头部信息到内存映射（mutable borrow）
    let header_data = &mut mmap[0..32];
    header_data[0] = updated_header.file_type;
    header_data[1] = updated_header.year;
    header_data[2] = updated_header.month;
    header_data[3] = updated_header.day;
    header_data[4..8].copy_from_slice(&updated_header.record_count.to_le_bytes());
    header_data[8..10].copy_from_slice(&updated_header.header_len.to_le_bytes());
    header_data[10..12].copy_from_slice(&updated_header.record_len.to_le_bytes());
    header_data[12..32].copy_from_slice(&updated_header.reserved);

    // 更新字段描述符
    let fields_start = 32;
    for (i, field) in updated_header.fields.iter().enumerate() {
        let field_data_start = fields_start + i * 32;
        let field_data = &mut mmap[field_data_start..field_data_start + 32];

        // 写入字段名称
        let name_len = field.name.len().min(11);
        field_data[0..name_len].copy_from_slice(field.name.as_bytes());
        field_data[name_len] = 0; // Null terminator

        // 写入字段类型、长度和小数位数
        field_data[11] = field.field_type as u8;
        field_data[16] = field.length;
        field_data[17] = field.decimal;

        // 清零剩余字节
        for j in 12..32 {
            if j != 16 && j != 17 {
                field_data[j] = 0;
            }
        }
    }

    // 清空多余的字段描述符区域
    let old_fields_end = fields_start + header.fields.len() * 32;
    let new_fields_end = fields_start + updated_header.fields.len() * 32;
    if new_fields_end < old_fields_end {
        mmap[new_fields_end..old_fields_end].fill(0);
    }

    // 更新记录数据
    let data_start = header.header_len as usize;
    let record_size = header.record_len as usize;
    let updated_record_size = updated_header.record_len as usize;
    let total_records = header.record_count as usize;

    for record_idx in 0..total_records {
        let old_offset = data_start + record_idx * record_size;
        let new_offset = data_start + record_idx * updated_record_size;
    
        let mut pos_old = 1; // 跳过删除标记
        let mut pos_new = 1;
    
        // 拷贝保留字段的数据
        for field in &updated_header.fields {
            let old_field = header.fields.iter().find(|f| f.name == field.name).unwrap();
            let old_end = pos_old + old_field.length as usize;
            let new_end = pos_new + field.length as usize;
    
            // 使用临时缓冲区避免同时借用
            let mut temp_buffer = vec![0u8; (old_end - pos_old)];
            temp_buffer.copy_from_slice(&mmap[old_offset + pos_old..old_offset + old_end]);
    
            let new_data = &mut mmap[new_offset + pos_new..new_offset + new_end];
            new_data.copy_from_slice(&temp_buffer);
    
            pos_old = old_end;
            pos_new = new_end;
        }
    
        // 清空多余的空间
        if updated_record_size > pos_new {
            mmap[new_offset + pos_new..new_offset + updated_record_size].fill(b' ');
        }
    }

    // 刷新内存映射到文件
    mmap.flush().context("Failed to flush memory map")?;
    file.sync_all().context("Failed to sync file")?;

    Ok(())
}

/// 判断记录是否匹配查询条件
fn is_record_matching(record: &HashMap<String, String>, query: &HashMap<String, String>) -> bool {
    query.iter().all(|(field_name, value)| {
        match record.get(field_name) {
            Some(field_value) => field_value == value,
            None => false,
        }
    })
}

/// 获取满足条件的记录数量
///
/// # 参数
///
/// * `path` - DBF 文件的路径。
/// * `query` - 查询条件，键为字段名，值为字段值。
///
/// # 返回值
///
/// 返回匹配记录的数量或错误。
pub fn get_filtered_record_count(path: &str, query: &HashMap<String, String>) -> Result<u32> {
    let file = File::open(path).context("Failed to open DBF file")?;
    let mmap = unsafe { Mmap::map(&file).context("Failed to map DBF file into memory")? };
    
    // 获取头部数据并解析
    let header = parse_header(&mmap)?;

    // 计算并行任务数量和每个任务处理的记录数量
    let num_threads = num_cpus::get().min(header.record_count as usize / 1000 + 1);
    let records_per_thread = (header.record_count as usize + num_threads - 1) / num_threads;

    let mmap = Arc::new(mmap); // 使用 Arc 共享内存映射
    let base_offset = header.header_len as usize;
    let record_size = header.record_len as usize;
    let fields = Arc::new(header.fields.clone()); // 共享字段描述符

    // 使用 rayon 进行并行处理
    let filtered_count = (0..num_threads).into_par_iter().map(|i| {
        let start_idx = i * records_per_thread;
        let end_idx = (start_idx + records_per_thread).min(header.record_count as usize);

        let start_offset = base_offset + start_idx * record_size;
        let end_offset = base_offset + end_idx * record_size;

        let chunk = &mmap[start_offset..end_offset];
        let mut local_count = 0;

        for record_data in chunk.chunks(record_size) {
            // 跳过删除标记
            if record_data[0] == 0x2A {
                continue;
            }

            // 解析记录
            match parse_record(record_data, &fields) {
                Ok(record) => {
                    // 检查记录是否匹配查询条件
                    if is_record_matching(&record, query) {
                        local_count += 1;
                    }
                }
                Err(e) => {
                    eprintln!("Error parsing record: {:?}", e);
                }
            }
        }

        local_count // 返回本线程处理的匹配记录数
    }).sum::<u32>(); // 汇总所有线程的匹配记录数

    Ok(filtered_count)
}